package nacos

import (
	"fmt"
	"net/url"
	"strconv"

	log "github.com/sirupsen/logrus"

	"github.com/nacos-group/nacos-sdk-go/clients"
	"github.com/nacos-group/nacos-sdk-go/clients/config_client"
	"github.com/nacos-group/nacos-sdk-go/clients/naming_client"
	"github.com/nacos-group/nacos-sdk-go/common/constant"
	"github.com/nacos-group/nacos-sdk-go/vo"
	"github.com/prometheus/client_golang/prometheus"
)

type Client struct {
	configClient config_client.IConfigClient
	namingClient naming_client.INamingClient
	namespace    string
	channel      chan int
	count        int
}

// Count 计数器类型
var watchCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
	Subsystem: "nacos",
	Name:      "config_watch_total",
	Help:      "Total number of config watched.",
}, []string{"group", "data_id"})

func init() {
	prometheus.MustRegister(watchCounter)
}

func NewNacosClient(nodes []string, config constant.ClientConfig) (client *Client, err error) {
	var configClient config_client.IConfigClient
	servers := []constant.ServerConfig{}
	for _, key := range nodes {
		nacosUrl, _ := url.Parse(key)

		port, _ := strconv.Atoi(nacosUrl.Port())
		servers = append(servers, constant.ServerConfig{
			IpAddr: nacosUrl.Hostname(),
			Port:   uint64(port),
		})
	}
	configClient, err = clients.CreateConfigClient(map[string]interface{}{
		"serverConfigs": servers,
		"clientConfig": constant.ClientConfig{
			TimeoutMs:           10000,
			ListenInterval:      20000,
			NotLoadCacheAtStart: true,
			NamespaceId:         config.NamespaceId,
			AccessKey:           config.AccessKey,
			SecretKey:           config.SecretKey,
			Endpoint:            config.Endpoint,
			OpenKMS:             config.OpenKMS,
			RegionId:            config.RegionId,
			Username:            config.Username,
			Password:            config.Password,
			LogLevel:            config.LogLevel,
		},
	})

	namingClient, _ := clients.CreateNamingClient(map[string]interface{}{
		"serverConfigs": servers,
		"clientConfig": constant.ClientConfig{
			TimeoutMs:           10000,
			ListenInterval:      20000,
			NotLoadCacheAtStart: true,
			NamespaceId:         config.NamespaceId,
			AccessKey:           config.AccessKey,
			SecretKey:           config.SecretKey,
			Endpoint:            config.Endpoint,
			OpenKMS:             config.OpenKMS,
			RegionId:            config.RegionId,
			Username:            config.Username,
			Password:            config.Password,
			LogLevel:            config.LogLevel,
		},
	})
	client = &Client{configClient, namingClient, config.NamespaceId, make(chan int, 10), 0}

	return
}

func (client *Client) GetContent(dataId string, group string) (string, error) {
	return client.configClient.GetConfig(vo.ConfigParam{DataId: dataId, Group: group})
}

func (client *Client) WatchContent(dataId string, group string, f func(dataId string, group string, content string)) (int, error) {
	err := client.configClient.ListenConfig(vo.ConfigParam{
		DataId: dataId,
		Group:  group,
		OnChange: func(namespace, group, dataId, data string) {
			log.Info(fmt.Sprintf("Config namespace=%s, dataId=%s, group=%s has changed,data: %s", namespace, dataId, group, data))
			watchCounter.WithLabelValues(group, dataId).Inc()
			f(dataId, group, data)
			for i := 0; i < client.count; i++ {
				client.channel <- 1
			}
		},
	})
	if err != nil {
		return 0, err
	}
	return 1, nil
}
